package edu.hawaii.ics.pjdb.worker.command_runners;

import java.sql.Connection;
import java.sql.SQLException;

import org.apache.thrift.TException;

import edu.hawaii.ics.pjdb.types.Command;
import edu.hawaii.ics.pjdb.types.WorkerCommand;
import edu.hawaii.ics.pjdb.util.JdbcHelper;

/**
 * A command runner takes as input a {@link WorkerCommand} and does everything
 * that's asked of the worker in said worker command.
 * 
 * There should be at least one {@link CommandRunner} for each type of command
 * in the {@link Command} union.
 * 
 */
public abstract class CommandRunner {
    /**
     * Runs a worker command through a set of default {@link CommandRunner}s.
     */
    public static class Switch {

        public void execute(WorkerCommand workerCommand) throws TException,
                SQLException {

            CommandRunner runner;
            Command cmd = workerCommand.getCommand();

            if (cmd.isSetDbSourceCmd()) {
                Connection localDb = JdbcHelper
                        .getDerbyConnection(workerCommand.getDbPath());
                runner = new DbSourceRunner(localDb);
                runner.execute(workerCommand);
            } else if (cmd.isSetRunScriptCmd()) {
                Connection localDb = JdbcHelper
                        .getDerbyConnection(workerCommand.getDbPath());
                runner = new DbScriptRunner(localDb);
                runner.execute(workerCommand);
            } else if (cmd.isSetDbEnrichCmd()) {
                Connection localDb = JdbcHelper
                        .getDerbyConnection(workerCommand.getDbPath());
                runner = new DbEnrichCommandRunner(localDb);
                runner.execute(workerCommand);
            } else if (cmd.isSetDbSyncCmd()) {
                Connection localDb = JdbcHelper
                        .getDerbyConnection(workerCommand.getDbPath());
                runner = new DbSyncRunner(localDb);
                runner.execute(workerCommand);
            } else if (cmd.isSetDbSyncRowSetCmd()) {
                Connection localDb = JdbcHelper
                        .getDerbyConnection(workerCommand.getDbPath());
                runner = new DbSyncRowSetRunner(localDb);
                runner.execute(workerCommand);
            } else if (cmd.isSetCsvSourceCmd()) {
                runner = new CsvSourceRunner(null);
                runner.execute(workerCommand);
            } else if (cmd.isSetCsvPartitionCmd()) {
                runner = new CsvPartitionRunner(null);
                runner.execute(workerCommand);
            } else {
                throw new UnsupportedOperationException(
                        "Unsupported command type - "
                                + cmd.getClass().getSimpleName());
            }
        }
    }

    protected final Connection localDb;

    /**
     * 
     * @param localDb
     *            a JDBC connection to the local database.
     */
    public CommandRunner(Connection localDb) {
        this.localDb = localDb;
    }

    public abstract void execute(WorkerCommand cmd) throws TException,
            SQLException;
}
